/**
 * Copyright (C) 2017 Baidu, Inc. All Rights Reserved.
 */
package com.baidu.jprotobuf.pbrpc.transport;

import java.util.concurrent.ExecutorService;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.baidu.jprotobuf.pbrpc.server.RpcServiceRegistry;
import com.baidu.jprotobuf.pbrpc.transport.handler.RpcDataPackageCompressHandler;
import com.baidu.jprotobuf.pbrpc.transport.handler.RpcDataPackageDecoder;
import com.baidu.jprotobuf.pbrpc.transport.handler.RpcDataPackageEncoder;
import com.baidu.jprotobuf.pbrpc.transport.handler.RpcDataPackageUnCompressHandler;
import com.baidu.jprotobuf.pbrpc.transport.handler.RpcServerChannelIdleHandler;
import com.baidu.jprotobuf.pbrpc.transport.handler.RpcServiceHandler;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * RPC server channel handler factory.
 *
 * @author xiemalin
 * @since 1.0
 */
public class RpcServerPipelineInitializer extends ChannelInitializer<Channel> {

    /** decode handler. */
    private static final String DECODER = "decoder";

    /** The Constant UNCOMPRESS. */
    private static final String UNCOMPRESS = "uncompress";

    /** The Constant COMPRESS. */
    private static final String COMPRESS = "compress";

    /** The Constant RPC_SERVER_HANDLER. */
    private static final String RPC_SERVER_HANDLER = "rpc_handler";

    /** The log. */
    private static Logger LOG = Logger.getLogger(RpcServerPipelineInitializer.class.getName());

    /** The rpc service registry. */
    private final RpcServiceRegistry rpcServiceRegistry;

    /** The Constant RPC_CHANNEL_STATE_AWARE_HANDLER. */
    private static final String RPC_CHANNEL_STATE_AWARE_HANDLER = "RpcChannelStateAwareHandler";

    /** The Constant RPC_CHANNEL_IDLE_HANDLER. */
    private static final String RPC_CHANNEL_IDLE_HANDLER = "idel_channal_handler";

    /** The Constant SERVER_DATA_PACK. */
    private static final String SERVER_DATA_PACK = "server_data_pack";

    /** The rpc server options. */
    private final RpcServerOptions rpcServerOptions;

    /** The es. */
    private ExecutorService es;

    /** The exception catcher. */
    private ExceptionCatcher exceptionCatcher;

    /**
     * Sets the exception catcher.
     *
     * @param exceptionCatcher the new exception catcher
     */
    public void setExceptionCatcher(ExceptionCatcher exceptionCatcher) {
        this.exceptionCatcher = exceptionCatcher;
    }

    /**
     * Instantiates a new rpc server pipeline initializer.
     *
     * @param rpcServiceRegistry the rpc service registry
     * @param rpcServerOptions the rpc server options
     * @param es the es
     * @param exceptionCatcher the exception catcher
     */
    public RpcServerPipelineInitializer(RpcServiceRegistry rpcServiceRegistry, RpcServerOptions rpcServerOptions,
            ExecutorService es, ExceptionCatcher exceptionCatcher) {
        this.rpcServiceRegistry = rpcServiceRegistry;
        this.rpcServerOptions = rpcServerOptions;
        this.es = es;
        this.exceptionCatcher = exceptionCatcher;

    }

    /*
     * (non-Javadoc)
     * 
     * @see io.netty.channel.ChannelInitializer#initChannel(io.netty.channel.Channel)
     */
    @Override
    protected void initChannel(Channel ch) throws Exception {
        LOG.log(Level.FINE, "begin process RPC server handler");
        ChannelPipeline channelPipe = ch.pipeline();
        // receive request data
        channelPipe.addLast(RPC_CHANNEL_STATE_AWARE_HANDLER,
                new IdleStateHandler(this.rpcServerOptions.getReaderIdleTime(),
                        this.rpcServerOptions.getWriterIdleTime(), this.rpcServerOptions.getKeepAliveTime()));

        channelPipe.addLast(RPC_CHANNEL_IDLE_HANDLER,
                new RpcServerChannelIdleHandler(this.rpcServerOptions.getKeepAliveTime()));

        int messageLengthFieldStart = 4;
        int messageLengthFieldWidth = 4;
        // Head meta size is 12, so messageLengthFieldStart +
        // messageLengthFieldWidth + adjustSize = 12;
        int adjustSize = 4;
        channelPipe.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(rpcServerOptions.getMaxSize(),
                messageLengthFieldStart, messageLengthFieldWidth, adjustSize, 0));

        // receive byte array to encode to RpcDataPackage

        RpcDataPackageDecoder rpcDataPackageDecoder =
                new RpcDataPackageDecoder(this.rpcServerOptions.getChunkPackageTimeout());
        channelPipe.addLast(DECODER, rpcDataPackageDecoder);
        // do uncompress handle
        channelPipe.addLast(UNCOMPRESS, new RpcDataPackageUnCompressHandler());
        // to process RPC service handler of request object RpcDataPackage and
        // return new RpcDataPackage
        RpcServiceHandler rpcServiceHandler = new RpcServiceHandler(this.rpcServiceRegistry, exceptionCatcher);
        rpcServiceHandler.setEs(es);
        channelPipe.addLast(RPC_SERVER_HANDLER, rpcServiceHandler);

        // response back
        // check if need to compress for data and attachment
        channelPipe.addFirst(COMPRESS, new RpcDataPackageCompressHandler());
        // encode RpcDataPackage to byte array
        channelPipe.addFirst(SERVER_DATA_PACK, new RpcDataPackageEncoder());

    }

    /**
     * Close.
     */
    public void close() {
        RpcDataPackageDecoder rpcDataPackageDecoder =
                new RpcDataPackageDecoder(this.rpcServerOptions.getChunkPackageTimeout());
        // to close the static thread pool
        rpcDataPackageDecoder.close();
    }

}
